
#include "task.h"

namespace datashuttle
{


BaseTask::BaseTask(IoLoop& loop)
	:mLoop(loop),mTargetTime(0),mDeltaTime(0),mTimerRowId( (size_t)-1),mTimerColId( (size_t)-1)
{
}

void BaseTask::setTimer( long long tte )
{
	if ( tte < 0 ) tte =0;
	mDeltaTime = tte;
	mTargetTime = timenow() + tte;
	mLoop.addTask(this);
}
void BaseTask::killTimer()
{
	if(mTimerRowId == (size_t)-1 || mTimerColId == (size_t)-1 )
		return;
	mLoop.removeTask(this);
}

///////////////////////////////////
///EventTask
EventTask::EventTask(IoLoop& loop)
	:BaseTask(loop),mFd(-1)
{
}

EventTask::~EventTask()
{
}


bool EventTask::addToIoLoop()
{
	if(!mLoop.addFd(this))
		return false;
	mSelf = this;
	return true;
}
void EventTask::removeFromIoLoop()
{
	if( mFd >= 0 )
		mLoop.removeFd(this);
	mSelf = 0;
}


////////////////////////////////////////
//AddrInfoHelper
class AddrInfoHelper
{
public:
	AddrInfoHelper()
	{
		memset(&mHints, 0, sizeof(mHints) );
		mInfo = 0;
		mHints.ai_family	= AF_UNSPEC;
		mHints.ai_socktype	= SOCK_STREAM;
		mHints.ai_protocol	= 0;
		mHints.ai_flags		= AI_CANONNAME|AI_PASSIVE;

	}
	virtual ~AddrInfoHelper()
	{
		freeResource();
	}

	addrinfo*	getinfo(const std::string& addr="", const std::string& port="" )
	{
		if( addr.empty() && port.empty())
			return mInfo;
		freeResource();
		int rc = getaddrinfo( addr.c_str() , port.c_str(), &mHints, &mInfo);
		if( rc != 0 )
		{
			mInfo = 0 ;
		}
		return mInfo;
	}
protected:
	void	freeResource()
	{
		if( mInfo)
		{
			freeaddrinfo( mInfo );
			mInfo = 0;
		}
	}

private:
	struct addrinfo mHints;
	struct addrinfo* mInfo;
};


////////////////////////////////////////
//Communicator
Communicator::Communicator()
	:mSock(-1)
{
}
Communicator::Communicator( int sock )
	:mSock(sock)
{
}
Communicator::~Communicator()
{
	shutdown();
}

bool Communicator::create( int family )
{
	shutdown();
	mSock = ::socket( family, SOCK_STREAM, 0 );
	if( mSock < 0 )
		return false;
	return true;
}

void Communicator::shutdown()
{
	if (mSock < 0 )
		return;
	::shutdown( mSock, SHUT_RDWR);// 我还确定这样将两个操作放在一起是否合适
	::close(mSock);
	mSock = -1;
}

bool Communicator::bind( const addrinfo* info)
{
	if( !info )
		return false;//failed to convert address/port pair the correct addrinfo format
	int rc = ::bind( mSock , info->ai_addr , (int)info->ai_addrlen );
	if( rc < 0 )
		return false;//log something here
	return true;

}
bool Communicator::bind( const std::string& address, const std::string& port )
{
	if( mSock < 0 )  return false;
	AddrInfoHelper aih;
	addrinfo* info = aih.getinfo( address, port );
	return bind(info);
}

bool Communicator::listen( int backlog )
{
	if( mSock < 0 )
		return false;
	int rc = ::listen( mSock, backlog );
	return rc == 0 ;
}

bool Communicator::connect( const addrinfo* info)
{
	if(!info)
		return false;
	int rc = ::connect( mSock, info->ai_addr, info->ai_addrlen );
	return rc == 0 ;
}

bool Communicator::connect( const std::string& address, const std::string& port )
{
	if( mSock < 0 || address.empty() || port.empty() )
		return false;
	AddrInfoHelper aih;
	addrinfo* info = aih.getinfo( address, port );
	return connect( info );
}

Communicator* Communicator::accept()
{
	if( mSock < 0 )		return NULL;
	int rc = ::accept( mSock, NULL , NULL );
	if( rc < 0 )
		return NULL;
	return new Communicator(rc);
}

bool Communicator::setNonBlock()
{
	if( mSock < 0 ) return false;
	int nFlags = fcntl(mSock, F_GETFL, 0);
	nFlags |= O_NONBLOCK;
	return fcntl( mSock, F_SETFL , nFlags);
}
bool Communicator::setBlock()
{
	if( mSock < 0 ) return false;
	int nFlags = fcntl(mSock, F_GETFL, 0);
	nFlags &= ~O_NONBLOCK;
	return fcntl( mSock, F_SETFL , nFlags);
}
bool Communicator::setWriteBuf( size_t bufSize )
{
	if( mSock < 0 )  return false;
	int rc = setsockopt( mSock, SOL_SOCKET, SO_SNDBUF, (char*)&bufSize , sizeof(bufSize) );
	return rc == 0;
}
bool Communicator::setReadBuf( size_t bufSize )
{
	if( mSock < 0 ) return false;
	int rc = setsockopt( mSock, SOL_SOCKET, SO_RCVBUF, (char*)&bufSize , sizeof(bufSize) );
	return rc == 0;	
}
bool Communicator::setWriteTimeo( size_t toms)
{
	if( mSock < 0 ) return false;
	int rc = setsockopt( mSock, SOL_SOCKET, SO_SNDTIMEO, (char*)&toms , sizeof(toms) );
	return rc == 0 ;
}
bool Communicator::setReadTimeo( size_t toms)
{
	if( mSock < 0 ) return false;
	int rc = setsockopt( mSock, SOL_SOCKET, SO_RCVTIMEO, (char*)toms, sizeof(toms) );
	return rc == 0 ;
}

int Communicator::read( char* buf , size_t size )
{
	if( mSock < 0)
		return COMM_ERROR_NOTSOCK;
	int rc = ::recv( mSock , buf, size , 0 );
	if ( rc < 0 )
	{
		return errcode2status(errno);
	}
	else
	{
		return rc;
	}
}
int Communicator::write( const char* buf, size_t size)
{
	if( mSock < 0 )
		return COMM_ERROR_NOTSOCK;
	int rc = ::send( mSock, buf, size, 0 );
	if( rc < 0 )
	{
		return errcode2status(errno);
	}
	else
	{
		return rc;
	}
}

COMM_ERROR Communicator::errcode2status( int err )
{
	COMM_ERROR status = COMM_ERROR_OK;
	switch( err )
	{
	case 0:			status = COMM_ERROR_OK; break;
	case EAGAIN:	status = COMM_ERROR_AGAIN; break;
	case ENOTSOCK:	status = COMM_ERROR_NOTSOCK; break;
	default:		status = COMM_ERROR_ERROR; break;
	}
	return status;
}

COMM_ERROR Communicator::lastError() const
{
	return errcode2status(errno);
}
bool getsocknameinfo( const struct sockaddr_storage* addr, size_t addrsize, std::string& ip, std::string& port)
{
	char bufhost[256]; bufhost[sizeof(bufhost)-1] = 0;
	char bufport[128]; bufport[sizeof(bufport)-1] = 0;
	
	int rc = getnameinfo((struct sockaddr*)&addr, sizeof(addr), 
			bufhost, sizeof(bufhost)-1, 
			bufport, sizeof(bufport)-1,
			NI_NUMERICHOST|NI_NUMERICSERV);
	if( rc != 0 )
		return false;
	ip = bufhost;
	port = bufport;
	return true;
}

bool Communicator::getLocalInfo( std::string& ip, std::string& port ) const
{
	if( mSock < 0 )
		return false;
	sockaddr_storage addr;
	memset(&addr,0,sizeof(addr));
	size_t sz = sizeof(addr);
	int rc = getsockname( mSock, (struct sockaddr*)&addr, &sz ) ;
	if( rc != 0 )
		return false;
	return getsocknameinfo( &addr, sz , ip, port );
}

bool Communicator::getPeerInfo( std::string& ip, std::string& port ) const
{
	if( mSock < 0 )
		return mSock;
	sockaddr_storage addr;
	memset(&addr,0,sizeof(addr));
	size_t sz = sizeof(addr);
	int rc = getpeername( mSock, (struct sockaddr*)&addr, &sz ) ;
	if( rc != 0 )
		return false;
	return getsocknameinfo( &addr, sz, ip, port );
}

/////////////////////////////////////////////////////////////
//CommunicationTask

CommunicationTask::CommunicationTask( IoLoop& loop )
	:EventTask(loop), mComm(0),mState(COMM_STATE_INIT)
{
}

CommunicationTask::~CommunicationTask()
{
	clearResource();
}

void CommunicationTask::clearResource()
{
	if( !mComm )
		return;
	mComm->shutdown();
	delete mComm;
	mComm = NULL;
	updateState( COMM_STATE_CLOSED );
}

void CommunicationTask::close()
{
	clearResource();
	if( getFd() >= 0 )
	{
		onClosed();
	}
	updateFd(-1);
	removeFromIoLoop();
}

bool CommunicationTask::listenAt( const std::string& address, const std::string& port , int backlog)
{
	AddrInfoHelper aih;
	struct addrinfo* info = aih.getinfo( address, port );
	if(!info)
	{//should I record some log here
		return false;
	}
	close();//关掉以前的instance
	mComm = new Communicator();
	mComm->setNonBlock();
	if( !mComm->create( info->ai_family ) )
		return false;
	if( !mComm->bind( info ) )
		return false;
	if( !mComm->listen( backlog ) )
		return false;
	
	updateFd( mComm->getSock() );
	addToIoLoop();//add current EventTask into IoLoop

	updateState( COMM_STATE_CREATED );

	onCreated();

	return true;
}

bool CommunicationTask::connectTo( const std::string& peerAddress, const std::string& peerPort, const std::string& localAddr, const std::string& localPort)
{
	close();
	mComm = new Communicator();
	mComm->setNonBlock();

	AddrInfoHelper aihPeer;
	struct addrinfo* peerinfo = aihPeer.getinfo( peerAddress , peerPort );
	
	if( !peerinfo)
		return false;
	if(!mComm->create( peerinfo->ai_family))
		return false;

	if( !localAddr.empty() && !localPort.empty())
	{
		AddrInfoHelper aihLocal;
		struct addrinfo* localinfo = aihLocal.getinfo(localAddr, localPort);
		if(!localinfo)
			return false;
		if(!mComm->bind(localinfo))
			return false;
	}
	
	if(!mComm->connect(peerinfo) && mComm->errcode2status(errno) != Communicator::COMM_STATUS_AGAIN )
		return false;

	updateFd( mComm->getSock());

	addToIoLoop();

	return true;
}

bool CommunicationTask::createFrom( Communicator* comm)
{
	if( !comm )
		return false;

	mComm = comm;
	mComm->setNonBlock();
	updateFd( mComm->getSock() );
	addToIoLoop();
	onCreated();
	return true;	
}

bool CommunicationTask::isValid() const
{
	return mComm && mComm->getSock() >= 0;//总感觉这样的判断还是不够好
}

void CommunicationTask::cancel()
{
	close();
}

void CommunicationTask::run( int type )
{
	if( type & TYPE_TASK_DATAIN )
	{
		canRead();
	}
	if( type & TYPE_TASK_DATAOUT )
	{
		if( mState < COMM_STATE_CREATED )
		{
			updateState( COMM_STATE_CREATED );
			onCreated();///communicator established
		}
		else
		{
			canWrite();
		}
	}
	if( type & TYPE_TASK_ERROR )
	{
		onError();
	}
	if( type & TYPE_TASK_TIMER )
	{
		resetTimerTablePosition();
		onTimer();
	}
}

COMM_ERROR CommunicationTask::lastError() const
{
   if(!mComm) 
	   return communicator::COMM_ERROR_NOTSOCK; 
   return mComm->lastError();
}
///////////////////////////////////////////////////////////////////
///
CommunicationServer::CommunicationServer( IoLoop& loop)
	:CommunicationTask(loop)
{
}
CommunicationServer::~CommunicationServer()
{
}
void CommunicationServer::canRead()
{
	Communicator* comm = 0;
	do
	{
		comm = getComm()->accept();
		if(!comm)
		{
			onError();
		}
		else
		{
			CommunicationTaskPtr task = new CommunicationTask( mLoop );
			if( !task->createFrom(comm) )
				task->close();
		}
	}while( comm != 0);
}
void CommunicationServer::onError()
{
	switch( getComm()->errcode2status(errno) )
	{
	case Communicator::COMM_STATUS_OK://do nothing
		break;
	case Communicator::COMM_STATUS_AGAIN:
		break;
	case Communicator::COMM_STATUS_ERROR:
		{//TODO: should I recreate the listen socket ?

		}
		break;
	case Communicator::COMM_STATUS_NOTSOCK:
		break;
	default:
		break;
	}
}
void CommunicationServer::onCancel()
{
	//I am not sure if I should close the communicator here
	close();
}


}//namespace datashuttle

